package com.digiwin.demo.job;

import com.digiwin.demo.util.MongoUtil;
import com.digiwin.demo.util.MysqlUtil;
import com.mongodb.hadoop.io.BSONWritable;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.types.Row;
import org.bson.BSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author yanggld
 * @Date 2019/12/26-14:14
 * @Description
 */
public class MongoToMysqlBatchJob {
	public static final String sql = "insert into u_sum (uid,sum) values (?,?)";
	public static void main(String[] args) throws Exception {
		//创建运行环境
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		//将mongo数据转化为Hadoop数据格式
		MongoUtil mongoUtil = new MongoUtil("test","book");
		DataSource<Tuple2<BSONWritable, BSONWritable>> input = mongoUtil.getInput(env);
		// 转换算子
		DataSet<Row> dataSet = input.map((MapFunction<Tuple2<BSONWritable, BSONWritable>, Row>) value -> {
			BSONWritable f1 = value.f1;
			BSONObject doc = f1.getDoc();
			String uid = String.valueOf(doc.get("uid"));
			Integer count = Integer.parseInt(doc.get("count").toString());
			Row row = new Row(2);
			row.setField(0, uid.getBytes("UTF-8"));
			row.setField(1, count);
			return row;
		}).groupBy((KeySelector<Row, String>) value -> {
			byte[] uid = (byte[])value.getField(0);
			return new String(uid);
		}).reduce((ReduceFunction<Row>) (oldValue, newValue) -> {
			int i1 = Integer.parseInt(oldValue.getField(1).toString());
			int i2 = Integer.parseInt(newValue.getField(1).toString());
			oldValue.setField(1, i1 + i2);
			return oldValue;
		});
		//将Row对象写到mysql
		dataSet.output(MysqlUtil.buildJDBCOutputFormat(sql));
		env.execute(Thread.currentThread().getStackTrace()[1].getClassName());
	}
}
